
package org.sopt.dev.ImgKnight.frontier.Queue;

import java.net.URL;
import java.security.InvalidParameterException;
import java.util.Collection;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.net.MalformedURLException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.log4j.Logger;

import org.sopt.dev.ImgKnight.frontier.Frontier;
import org.sopt.dev.ImgKnight.frontier.Database.ToVisitDAO;
import org.sopt.dev.ImgKnight.frontier.Filter.ExternalUrlFilter;
import org.sopt.dev.ImgKnight.frontier.Filter.GeoLocationFilter;
import org.sopt.dev.ImgKnight.frontier.Filter.RabinHashFunction;
import org.sopt.dev.ImgKnight.frontier.Filter.UrlDupFilter;
import org.sopt.dev.ImgKnight.frontier.Filter.UrlPatternFilter;

public class SiteMapHashTable {

	private Frontier frontier = null;
	
	private GeoLocationFilter geoFilter = null;
	
	private RabinHashFunction fp = null;
	
	private ExternalUrlFilter externalUrlFilter = null;
	
	private UrlPatternFilter urlPatternFilter = null;
	
	private UrlDupFilter urlDupFilter = null;
	
	private ConcurrentHashMap<Long, SiteMapList> highPrioSiteHashMap = null;
	
	private ConcurrentHashMap<Long, SiteMapList> SiteHashMap = null;
	
	private int cacheHighWaterMark = -1;
	
	private ConcurrentLinkedQueue<String> urlCache = null; 

	private int entryLowWaterMark = -1;
	
	private int entryFetchSize = -1;
	
	private ToVisitDAO toVisitDAO = null;
	
	private AtomicLong duplicatedUrlCount = null;
	
	private AtomicLong DiscardedUrlCount = null;
	
	private AtomicLong popedCount = null;
	
	private AtomicLong highPrioMemPushedCount = null;
	
	private AtomicLong memPushedCount = null;
	
	private AtomicLong flushDiskCount = null;
	
	boolean toVisitFlushed = false;
	
	private Logger serviceLogger = null;
	
	private Logger exceptionLogger = null;
	
	public SiteMapHashTable(String[] seedUrl, 
			int cacheHighWaterMark, int entryLowWaterMark, int entryFetchSize, 
			boolean isCleanStart) {
		frontier = Frontier.getInstance();
		geoFilter = new GeoLocationFilter(65536);
		fp = new RabinHashFunction();
		externalUrlFilter = new ExternalUrlFilter();
		urlPatternFilter = new UrlPatternFilter();
		urlDupFilter = new UrlDupFilter(frontier.getFrontierConfig().FPdocsHighWaterMark, isCleanStart);
		highPrioSiteHashMap = new ConcurrentHashMap<Long, SiteMapList>();
		SiteHashMap = new ConcurrentHashMap<Long, SiteMapList>();
		urlCache = new ConcurrentLinkedQueue<String>();
		this.cacheHighWaterMark = cacheHighWaterMark;
		this.entryLowWaterMark = entryLowWaterMark;
		this.entryFetchSize = entryFetchSize;
		
		toVisitDAO = new ToVisitDAO(isCleanStart);
		
		duplicatedUrlCount = new AtomicLong(0);
		DiscardedUrlCount = new AtomicLong(0);
		popedCount = new AtomicLong(0);
		highPrioMemPushedCount = new AtomicLong(0);
		memPushedCount = new AtomicLong(0);
		flushDiskCount = new AtomicLong(toVisitDAO.selectTovisitSize());
		
		URL url = null;
		Long hostHashValue = null;
		Long urlHashValue = null;
		SiteMapList siteMapList = null;
		for(String seed : seedUrl) {
			try {
				url = new URL(seed);
				hostHashValue = fp.hash(url.getHost());
				urlHashValue = fp.hash(url.toString());
				if( !urlDupFilter.seenURL(url, hostHashValue, urlHashValue) ) {
					String countryCode = geoFilter.getGeoCountryCode(url);
					String countryName = geoFilter.getGeoCountryName(url);
					siteMapList = new SiteMapList(hostHashValue, countryCode, countryName);
					siteMapList.offer(url);
					SiteHashMap.put(hostHashValue, siteMapList);
					memPushedCount.incrementAndGet();
				}
			} catch (MalformedURLException e) {
			} catch (IllegalArgumentException e) {
			}
		}
		if( flushDiskCount.longValue() > 0 )
			toVisitFlushed = true;
		serviceLogger = Logger.getLogger("frontier.service");
		exceptionLogger = Logger.getLogger("frontier.exception");
	}

	public void reserveTracking(String seed) {
		URL url = null;
		try {
			url = new URL(seed);
		} catch (MalformedURLException e) {
			return;
		}
		Long hostHashValue = fp.hash(url.getHost());
		Long urlHashValue = fp.hash(url.toString());
		if( !urlDupFilter.seenURL(url, hostHashValue, urlHashValue) ) {
			if( !frontier.getFrontierConfig().FetchExternalUrl ) {
				frontier.getFrontierConfig().addSeedUrl(seed);
			}
			SiteMapList siteMapList = SiteHashMap.get(hostHashValue);
			if( siteMapList != null ) {
				siteMapList.offer(url);
			}
			else {
				String countryCode = geoFilter.getGeoCountryCode(url);
				String countryName = geoFilter.getGeoCountryName(url);
				if( countryCode == null || countryName == null )
					return;
				siteMapList = new SiteMapList(hostHashValue, countryCode, countryName);
				siteMapList.offer(url);
				SiteHashMap.put(hostHashValue, siteMapList);
			}
			serviceLogger.info("Tracking reserved: "+url);
			memPushedCount.incrementAndGet();
		}
	}
	
	public int offerToHighPrioMem(List<String> extractedUrl) {
		int pushedCount = 0;
		URL url = null;
		Long hostHashValue = null;
		SiteMapList siteMapList = null;
		for(String extract : extractedUrl) {
			try {
				if( extract.startsWith("http://") ) {
					url = new URL(extract);
					hostHashValue = fp.hash(url.getHost());
					
					siteMapList = highPrioSiteHashMap.get(hostHashValue);
					if( siteMapList != null ) {
						siteMapList.offer(url);
					}
					else {
						String countryCode = geoFilter.getGeoCountryCode(url);
						String countryName = geoFilter.getGeoCountryName(url);
						if( countryCode == null || countryName == null )
							continue;
						siteMapList = new SiteMapList(hostHashValue, countryCode, countryName);
						siteMapList.offer(url);
						highPrioSiteHashMap.put(hostHashValue, siteMapList);
					}
					pushedCount++;
				}
			} catch(MalformedURLException e) {
				exceptionLogger.warn("MalformedURLException: "+e);
			} catch (InvalidParameterException e) {
				exceptionLogger.debug("InvalidParameterException: "+e);
			} catch(Exception e) {
				exceptionLogger.warn("Unexpected exception: "+e);
				e.printStackTrace();
			}
		}
		highPrioMemPushedCount.addAndGet(pushedCount);
		return pushedCount;
	}
	
	private int offerToMem(List<String> extractedUrl) {
		int pushedCount = 0;
		URL url = null;
		Long hostHashValue = null;
		SiteMapList siteMapList = null;
		for(String extract : extractedUrl) {
			try {
				if( extract.startsWith("http://") ) {
					url = new URL(extract);
					hostHashValue = fp.hash(url.getHost());
					
					siteMapList = SiteHashMap.get(hostHashValue);
					if( siteMapList != null ) {
						siteMapList.offer(url);
					}
					else {
						String countryCode = geoFilter.getGeoCountryCode(url);
						String countryName = geoFilter.getGeoCountryName(url);
						if( countryCode == null || countryName == null )
							continue;
						siteMapList = new SiteMapList(hostHashValue, countryCode, countryName);
						siteMapList.offer(url);
						SiteHashMap.put(hostHashValue, siteMapList);
					}
					pushedCount++;
				}
			} catch(MalformedURLException e) {
				exceptionLogger.warn("MalformedURLException: "+e);
			} catch (InvalidParameterException e) {
				exceptionLogger.debug("InvalidParameterException: "+e);
			} catch(Exception e) {
				exceptionLogger.warn("Unexpected exception: "+e);
				e.printStackTrace();
			}
		}
		memPushedCount.addAndGet(pushedCount);
		return pushedCount;
	}
	
	public void memToDisk() {
		urlDupFilter.memToDisk();
	}
	
	public int offerToCache(List<String> extracted) {
		URL url = null;
		Long hostHashValue = null;
		Long urlHashValue = null;
		List<String> highprio = new Vector<String>(extracted.size());
		List<String> tovisit = new Vector<String>(extracted.size());
		for(String extract : extracted) {
			try {
				if( extract.startsWith("http://") ) {
					url = new URL(extract);
					if( !externalUrlFilter.validate(url) ) {
						//serviceLogger.info("Ignored(External Url): "+extract);
						DiscardedUrlCount.incrementAndGet();
						continue;
					}
					if( !urlPatternFilter.validate(url) ) {
						//serviceLogger.info("Ignored(Disallowd UrlPattern): "+extract);
						DiscardedUrlCount.incrementAndGet();
						continue;
					}
					if( !geoFilter.validate(url) ) {
						//serviceLogger.info("Ignored(Undefine GeoLocation): "+extract);
						DiscardedUrlCount.incrementAndGet();
						continue;
					}
					hostHashValue = fp.hash(url.getHost());
					urlHashValue = fp.hash(extract);
					if( !urlDupFilter.seenURL(url, hostHashValue, urlHashValue) ) {
						if( frontier.getFrontierConfig().containsHighPrioKeyword(extract) )
							highprio.add(extract);
						else
							tovisit.add(extract);
					}
					else {
						//serviceLogger.info("Ignored(aleady seen URL): "+extract);
						duplicatedUrlCount.incrementAndGet();
					}
				}
			} catch(MalformedURLException e) {
				//exceptionLogger.warn("MalformedURLException: "+e);
			} catch(Exception e) {
				exceptionLogger.warn("Unexpected exception: "+e);
				e.printStackTrace();
			}
		}
		
		int addCount = 0;
		if( getHighPrioMemPushedCount() < 
				frontier.getFrontierConfig().HighPrioSiteHashTableHighWaterMark )
			addCount = offerToHighPrioMem(highprio);
		else 
			tovisit.addAll(highprio);
		
		synchronized (this) {
			int cacheSize = urlCache.size();
			if( cacheSize >= cacheHighWaterMark )
				cacheToDisk();
			urlCache.addAll(tovisit);
		}
		addCount += tovisit.size();
		return addCount;
	}

	private void cacheToDisk() {
		int cacheSize = urlCache.size();
		String [] url = new String[cacheSize];
		urlCache.toArray(url);
		if( toVisitDAO.insertTovisit(url) ) {
			toVisitFlushed = true;
			flushDiskCount.addAndGet(cacheSize);
		}
		else {
			toVisitDAO.deleteTovisitByRabinHashCode(url);
			if( toVisitDAO.insertTovisit(url) ) {
				toVisitFlushed = true;
				flushDiskCount.addAndGet(cacheSize);
			}
			else {
				exceptionLogger.error("UrlCache flush fail");
			}
		}
		urlCache.clear();
	}
	
	@SuppressWarnings("unchecked")
	public Map take() {
		long politenessTimeMillis = frontier.getFrontierConfig().PolitenessTimeMillis;
		while( true ) {
			Map highPrioUrlMap = getHighPrioURL(politenessTimeMillis);
			if( highPrioUrlMap != null )
				return highPrioUrlMap;
			
			memFill();
			
			Collection<SiteMapList> collection = SiteHashMap.values();
			if( collection == null || collection.isEmpty() ) {
				try {
					Thread.sleep(politenessTimeMillis);
					continue;
				} catch (InterruptedException e) {
					Thread.currentThread().interrupt();
					return null;
				}
			}
			
			long now = System.currentTimeMillis();
			for(SiteMapList siteMapList : collection) {
				if( siteMapList.interval(now) > politenessTimeMillis ) {
					URL url = siteMapList.poll();
					if( siteMapList.isEmpty() ) {
						Long hostHashValue = siteMapList.getHostHashValue();
						SiteHashMap.remove(hostHashValue);
					}
					if( url != null ) {
						popedCount.incrementAndGet();
						memPushedCount.decrementAndGet();
						String countryCode = siteMapList.getCountryCode();
						String countryName = siteMapList.getCountryName();
						Map pair = new Hashtable();
						pair.put("countryCode", countryCode);
						pair.put("countryName", countryName);
						pair.put("URL", url);
						return pair;
					}
				}
			}
		}
	}

	@SuppressWarnings("unchecked")
	private Map getHighPrioURL(long politenessTimeMillis) {
		Map urlMap = null;
		Collection<SiteMapList> collection = highPrioSiteHashMap.values();
		if( collection != null && !collection.isEmpty() ) {
			long now = System.currentTimeMillis();
			for(SiteMapList siteMapList : collection) {
				if( siteMapList.interval(now) > politenessTimeMillis ) {
					URL url = siteMapList.poll();
					if( siteMapList.isEmpty() ) {
						Long hostHashValue = siteMapList.getHostHashValue();
						highPrioSiteHashMap.remove(hostHashValue);
					}
					if( url != null ) {
						popedCount.incrementAndGet();
						highPrioMemPushedCount.decrementAndGet();
						String countryCode = siteMapList.getCountryCode();
						String countryName = siteMapList.getCountryName();
						urlMap = new Hashtable();
						urlMap.put("countryCode", countryCode);
						urlMap.put("countryName", countryName);
						urlMap.put("URL", url);
					}
				}
			}
		}
		return urlMap;
	}
	
	private void memFill() {
		if( toVisitFlushed ) {
			if( getMemPushedCount() <= entryLowWaterMark ) {
				List<String> extractedFromDisk = toVisitDAO.selectTovisitAndDelete(entryFetchSize);
				offerToMem(extractedFromDisk);
				flushDiskCount.set(flushDiskCount.get()-extractedFromDisk.size());
				if( flushDiskCount.get() <= 0 )
					toVisitFlushed = false;
			}
		} 
		else {
			if( getMemPushedCount() <= entryLowWaterMark ) {
				synchronized (this) {
					int cacheSize = urlCache.size();
					if( cacheSize > 0 && cacheSize <= entryFetchSize ) {
						List<String> url = new Vector<String>(cacheSize);
						for(String u : urlCache) {
							url.add(u);
						}
						offerToMem(url);
						urlCache.clear();
					}
					else if( cacheSize > entryFetchSize ){
						List<String> url = new Vector<String>(entryFetchSize);
						for(int i=0; i<entryFetchSize; i++){
							url.add(urlCache.poll());
						}
						offerToMem(url);
					}
				}
			}
		}
	}
	
	public String[] getHighPrioUrl() {
		Collection<SiteMapList> collection = highPrioSiteHashMap.values();
		if( collection != null && !collection.isEmpty() ) {
			int i = 0;
			String [] highPrioUrl = new String[collection.size()];
			for(SiteMapList siteMapList : collection) {
				highPrioUrl[i++] = siteMapList.peek().getHost();
			}
			return highPrioUrl;
		}
		return null;
	}
	
	public synchronized int removeUrl(List<URL> removeUrlList, 
				boolean highPrio, boolean mem, boolean cache) {
		int rmCount = 0;
		for(URL url : removeUrlList) {
			Long hostHashValue = fp.hash(url.getHost());
			if( highPrio ) {
				SiteMapList siteMapLink = highPrioSiteHashMap.get(hostHashValue);
				if( siteMapLink != null ) {
					highPrioSiteHashMap.remove(hostHashValue);
					highPrioMemPushedCount.set(getHighPrioMemPushedCount()-siteMapLink.size());
					rmCount = siteMapLink.size();
					serviceLogger.info("Url removed from highPrioSiteHashMap: "+rmCount);
				}
			}
			if( mem ) {
				SiteMapList siteMapLink = SiteHashMap.get(hostHashValue);
				if( siteMapLink != null ) {
					SiteHashMap.remove(hostHashValue);
					memPushedCount.set(getMemPushedCount()-siteMapLink.size());
					serviceLogger.info("Url removed from siteHashMap: "+siteMapLink.size());
					rmCount += siteMapLink.size();
				}
			}
			if( cache ) {
				int rmCountl = 0;
				for(Iterator<String> it=urlCache.iterator(); it.hasNext(); /**/) {
					String c = it.next();
					if( c.startsWith(url.toString()) ) {
						urlCache.remove(c);
						++rmCount;
						++rmCountl;
					}
				}
				if( rmCount > 0 )
					serviceLogger.info("Url removed from UrlCache: "+rmCountl);
			}
		}
		return rmCount;
	}
	
	public int getHighPrioSiteHashMapSize() {
		return highPrioSiteHashMap.size();
	}
	
	public int getSiteHashMapSize() {
		return SiteHashMap.size();
	}
	
	public int getEntryLowWaterMark() {
		return entryLowWaterMark;
	}

	public int getUrlDupFilterHighWaterMark() {
		return urlDupFilter.getFpdocsHighWaterMark();
	}
	
	public long getDuplicatedUrlCount() {
		return duplicatedUrlCount.get();
	}

	public long getDiscardedUrlCount() {
		return DiscardedUrlCount.get();
	}
	
	public long getPopedCount() {
		return popedCount.get();
	}
	
	public long getTotalPushedCount() {
		return getHighPrioMemPushedCount()+ 
		getMemPushedCount()+
		getCachePushedCount()+
		getTovisitPushedCount()+
		getPopedCount();
	}
	
	public long getHighPrioMemPushedCount() {
		return highPrioMemPushedCount.get();
	}
	
	public long getMemPushedCount() {
		return memPushedCount.get();
	}
	
	public long getCachePushedCount() {
		return urlCache.size();
	}
	
	public long getTovisitPushedCount() {
		return flushDiskCount.get();
	}
	
	public long getUrlDupFilterFlushedCount() {
		return urlDupFilter.getFpDocsFlushedCount();
	}
	
	public int getUrlDupFilterCurrentCount() {
		return urlDupFilter.getFpDocsCurrentCount();
	}
	
}
